package com.zheng.redpacket.server.config;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;

@Configuration
public class RabbitmqConfig {
    private static final Logger LOGGER = LoggerFactory.getLogger(RabbitmqConfig.class);
    //自动设置Rabbitmq连接工厂实例
    @Autowired
    private CachingConnectionFactory connectionFactory;
    //消息监听器所在容器工厂配置类实例
    @Autowired
    private SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer;
    //读取配置文件的环境变量实例
    @Autowired
    private Environment env;

    /**
     * 单一消费者实例配置
     * @return
     */
    @Bean(name = "singleListenerContainer")
    public SimpleRabbitListenerContainerFactory singleListenerContainer() {
        //定义消息监听器所在工厂类
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        //设置工厂实例
        factory.setConnectionFactory(connectionFactory);
        /**
         * 工厂配置
         * 1.设置消息传输格式为json
         * 2.设置并发消费者实例的初始数量为1个
         * 3.最大并发消费者实例数设为1个
         * 4.每个消费者实例拉取消息的数量设为1条
         */
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setConcurrentConsumers(1);
        factory.setMaxConcurrentConsumers(1);
        factory.setPrefetchCount(1);
        return factory;
    }

    @Bean(name = "multiListenerContainer")
    public SimpleRabbitListenerContainerFactory multiListenerContainer() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factoryConfigurer.configure(factory, connectionFactory);
        factory.setConnectionFactory(connectionFactory);
        /**
         * 工厂配置
         * 1.设置消息传输格式为json
         * 2.消息的确认消费模式设为NONE，表示不需要确认消费
         * 3.设置并发消费者实例的初始数量为10个
         * 4.最大并发消费者实例数设为15个
         * 5.每个消费者实例拉取消息的数量设为10条
         */
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setAcknowledgeMode(AcknowledgeMode.NONE);
        factory.setConcurrentConsumers(10);
        factory.setMaxConcurrentConsumers(15);
        factory.setPrefetchCount(10);
        return factory;
    }

    /**
     * 自定义发送消息组件RabbitTemplate
     * @return
     */
    @Bean
    public RabbitTemplate rabbitTemplate(){
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        connectionFactory.setPublisherReturns(true);
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                LOGGER.info("消息发送成功：correlationData({}), ack({}), cause({})", correlationData, ack, cause);
            }
        });
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String reply, String exchange, String routing) {
                LOGGER.info("消息丢失：exchange({}), routing({}), replyCode({}), reply({}), message:{}",exchange, routing, replyCode, reply, message);
            }
        });
        return rabbitTemplate;
    }

    /**
     * 创建队列
     * @return
     */
    @Bean(name = "basicQueue")
    public Queue basicQueue() {
        return new Queue(env.getProperty("mq.basic.info.queue.name"), true);
    }

    /**
     * 消息模型
     * @return
     */
    @Bean
    public DirectExchange basicExchange() {
        return new DirectExchange(env.getProperty("mq.basic.info.exchange.name"), true, false);
    }

    @Bean
    public Binding basicBinding() {
        return BindingBuilder.bind(basicQueue()).to(basicExchange()).with(env.getProperty("mq.basic.info.routing.key.name"));
    }


    @Bean(name = "loginQueue")
    public Queue loginQueue() {
        return new Queue(env.getProperty("mq.login.info.queue.name"), true);
    }

    @Bean
    public TopicExchange loginExchange() {
        return new TopicExchange(env.getProperty("mq.login.info.exchange.name"), true, false);
    }

    @Bean
    public Binding loginBinding() {
        return BindingBuilder.bind(loginQueue()).to(loginExchange()).with(env.getProperty("mq.login.info.routing.key.name"));
    }

    /**
     * 创建延时队列
     * @return
     */
    @Bean
    public Queue orderDelayQueue() {
        //添加延时队列组成成分
        Map<String, Object> args = new HashMap<>();
        args.put("x-dead-letter-exchange", env.getProperty("mq.order.dead.exchange.name"));
        args.put("x-dead-letter-routing-key", env.getProperty("mq.order.dead.routing.key.name"));
        args.put("x-message-ttl", 10000);
        //创建延时队列
        return new Queue(env.getProperty("mq.order.dead.queue.name"), true, false, false, args);
    }

    @Bean
    public TopicExchange orderDelayExchange() {
        return new TopicExchange(env.getProperty("mq.order.dead.exchange.name"), true, false);
    }

    @Bean
    public Binding orderDelayBinding() {
        return BindingBuilder.bind(orderDelayQueue()).to(orderProducerExchange()).with(env.getProperty("mq.producer.order.routing.key.name"));
    }

    /**
     * 创建真正队列
     */
    @Bean
    public Queue realOrderConsumerQueue() {
        return new Queue(env.getProperty("mq.consumer.order.queue.name"));
    }

    @Bean
    public TopicExchange orderProducerExchange() {
        return new TopicExchange(env.getProperty("mq.producer.order.exchange.name"), true, false);
    }

    @Bean
    public Binding orderProducerBinding() {
        return BindingBuilder.bind(realOrderConsumerQueue()).to(orderDelayExchange()).with(env.getProperty("mq.order.dead.routing.key.name"));
    }



}
